#include "zhelpers.h"
#include "proto.msg.pb.h"
#include "gf_thread.h"
#include "gf_server.h"
#include "Log.h"
#include "vs_opencv.h"
#include "GlobalFeature_API.h"

int gf_extract_thread::thread_func()
{
	if(get_thread_pool() == NULL){
		LOG_ERROR_VA("[%s] pool_work_thread don't belong to a thread_pool", get_name());
		return -1;
	}
	int ret;
	void *receiver = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REP);
	if(receiver == NULL){
		LOG_ERROR_VA("[%s] zmq_socket failed", get_name());
		return -1;
	}
	ret = zmq_connect(receiver, "tcp://199.155.122.11:45511"); //gf_extract back_address

	LOG_DEBUG_VA("[%s] thread start", get_name());
	while(!(get_thread_pool()->has_stoped())){
		int size;
		zmq_msg_t recv_msg;
		zmq_msg_init(&recv_msg);

		LOG_DEBUG_VA("[%s] thread get task...", get_name());
		size = zmq_recvmsg(receiver, &recv_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		string proto_buf;
		proto_buf.assign((char *)zmq_msg_data(&recv_msg), size);
		proto::extract_req extract_req;
		if(!extract_req.ParseFromString(proto_buf)){
			LOG_ERROR_VA("[%s] extract_req ParseFromString fail", get_name());
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("[%s] thread recv length: %d", get_name(), size);
		//the work of engine
		//LOG_DEBUG_VA("image width: %d, height: %d, depth: %d, channel: %d, size: %d", extract_req.image_width(), extract_req.image_height(), extract_req.image_depth(), extract_req.image_channel(), extract_req.image_data().size());
		char *image_data = new char[extract_req.image_data().size()];
		memcpy(image_data, extract_req.image_data().c_str(), extract_req.image_data().size());
		IplImage *ipl = cv_create_image(extract_req.image_width(), extract_req.image_height(), extract_req.image_depth(), extract_req.image_channel(), image_data, extract_req.image_data().size());
		delete [] image_data;
		if(ipl == NULL){
			LOG_ERROR_VA("[%d] open_cv load error", get_name());
			send_error_reply(receiver);
			continue;
		}

		int vid_num = 8;
		int vids[8] = {0};
		unsigned int feature_buf_size = GLOBALFEATURE_3_2_0::GetFeatureValueLength() + 9*sizeof(int);
		char feature_buf[feature_buf_size];

		ret = GLOBALFEATURE_3_2_0::Extraction(ipl, NULL, extract_req.class_id(), &vid_num, vids, feature_buf);
		cv_release_image(ipl);
		if(ret != 0){
			LOG_ERROR_VA("Global Extract failed: %d", ret);
			send_error_reply(receiver);
			continue;
		}

		//send resp to front
		proto::gf_extract_resp gf_extract_resp;
		gf_extract_resp.mutable_head()->set_version(420);
		gf_extract_resp.mutable_head()->set_command(420);
		gf_extract_resp.mutable_head()->set_password("xxxxxxxx", 8);
		gf_extract_resp.mutable_head()->set_type(1);
		gf_extract_resp.mutable_head()->set_status(0);
		gf_extract_resp.mutable_head()->set_src(1);
		gf_extract_resp.mutable_head()->set_dest(2);
		gf_extract_resp.mutable_head()->set_sn_high(123456789);
		gf_extract_resp.mutable_head()->set_sn_low(123456789);
		gf_extract_resp.mutable_head()->set_length(0);

		for(int i = 0; i < vid_num; i++){
			gf_extract_resp.add_vids(vids[i]);
		}
		gf_extract_resp.set_feature_data(feature_buf,feature_buf_size);
		
		proto_buf.clear();
		if(!gf_extract_resp.SerializePartialToString(&proto_buf)){
			LOG_ERROR_VA("SerializePartialToString fail");
			send_error_reply(receiver);
			continue;
		}
		zmq_msg_t gf_extract_resp_msg;
		zmq_msg_init_size(&gf_extract_resp_msg, proto_buf.size());
		memcpy(zmq_msg_data(&gf_extract_resp_msg), proto_buf.c_str(), proto_buf.size());
		size = zmq_sendmsg(receiver, &gf_extract_resp_msg, 0);
		if(size == -1){
			send_error_reply(receiver);
			LOG_ERROR_VA("zmq_sendmsg error");
			continue;
		}
	}
	zmq_close(receiver);
}

int gf_merge_thread::thread_func(){
	if(get_thread_pool() == NULL){
		LOG_ERROR_VA("[%s] pool_work_thread don't belong to a thread_pool", get_name());
		return -1;
	}
	int ret;
	void *receiver = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REP);
	if(receiver == NULL){
		LOG_ERROR_VA("[%s] zmq_socket failed", get_name());
		return -1;
	}
	ret = zmq_connect(receiver, "tcp://199.155.122.11:45512"); //gf_merge back_address

	LOG_DEBUG_VA("[%s] thread start...", get_name());
	while(!(get_thread_pool()->has_stoped())){
		int size;
		zmq_msg_t recv_msg;
		zmq_msg_init(&recv_msg);

		LOG_DEBUG_VA("[%s] thread get task...", get_name());
		size = zmq_recvmsg(receiver, &recv_msg, 0);
		if(size == -1){
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		string proto_buf;
		proto_buf.assign((char *)zmq_msg_data(&recv_msg), size);
		proto::gf_match_req gf_match_req;
		if(!gf_match_req.ParseFromString(proto_buf)){
			LOG_ERROR_VA("[%s] gf_match_req ParseFromString fail", get_name());
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("[%s] recv length: %d", get_name(), size);
		//send to all match
		int send_match_req_num = 0;
		string gf_match_tcp_addr;
		vector<string> alive_gf_match_addr_vec;
		vector<void *> alive_gf_match_vec;
		for(vector<address>::iterator it = get_gf_match().begin(); it != get_gf_match().end(); it++){
			//connect every time
			//not a good idea
			gf_match_tcp_addr.clear();
			gf_match_tcp_addr.append("tcp://");
			gf_match_tcp_addr.append(it->get_ip());
			gf_match_tcp_addr.append(":");
			gf_match_tcp_addr.append(it->get_port());
			void *zmq_gf_match = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REQ);
			if(zmq_gf_match == NULL){
				LOG_ERROR_VA("[%s] match_zmq zmq_socket failed", get_name());
				return -1;
			}
			ret = zmq_connect(zmq_gf_match, gf_match_tcp_addr.c_str()); //gf_match front_address
			if(ret == -1){
				LOG_ERROR_VA("zmq_connect error");
				continue;
			}
			zmq_msg_t match_req_msg;
			zmq_msg_init_size(&match_req_msg, size);
			memcpy(zmq_msg_data(&match_req_msg), zmq_msg_data(&recv_msg), size);
			//zmq_msg_copy(&match_req_msg, &recv_msg);
			int size = zmq_sendmsg(zmq_gf_match, &match_req_msg, 0);
			if(size == -1){
				LOG_ERROR_VA("zmq_sendmsg error");
				continue;
			}
			alive_gf_match_vec.push_back(zmq_gf_match);
			send_match_req_num++;
		}
		//send over, start to receive for all matchs
		if(send_match_req_num == 0){
			LOG_ERROR_VA("no match send success");
			send_error_reply(receiver);
			continue;
		}
		//zmq_poll recv
		int recv_match_req_num = 0;
		vector<zmq_msg_t> gf_match_resp_msg_vec;
		vector<proto::gf_match_resp> gf_match_resp_vec;
		while(recv_match_req_num < send_match_req_num){
			//the timeout!!!
			LOG_DEBUG_VA("[%s] prepare to recv from %d match", get_name(), send_match_req_num);
			//zmq_pollitem_t *poll_items = new zmq_pollitem_t[alive_gf_match_vec.size()];
			zmq_pollitem_t poll_items[alive_gf_match_vec.size()];
			for(int i = 0; i < alive_gf_match_vec.size(); i++){
				poll_items[i].socket = alive_gf_match_vec[i];
				poll_items[i].fd = 0;
				poll_items[i].events = ZMQ_POLLIN;
				poll_items[i].revents = 0;
			}
			int zmq_poll_ret = zmq_poll(poll_items, alive_gf_match_vec.size(), -1);
			if(zmq_poll_ret < 0)
				LOG_ERROR_VA("[%s] zmq_poll error");
			int index = 0;
			int alive_num = alive_gf_match_vec.size();
			for(int i = 0; i < alive_num; i++){
				if(poll_items[i].revents & ZMQ_POLLIN){
					LOG_DEBUG_VA("[%s] get a ZMQ_POLLIN", get_name());
					zmq_msg_t gf_match_resp_msg;
					zmq_msg_init(&gf_match_resp_msg);
					zmq_msg_recv(&gf_match_resp_msg, poll_items[i].socket, 0);
					//if recv sucess, erase if from alive_gf_match_vec!!!
					proto_buf.clear();
					proto_buf.assign((char *)zmq_msg_data(&gf_match_resp_msg),zmq_msg_size(&gf_match_resp_msg));
					proto::gf_match_resp gf_match_resp;
					if(!gf_match_resp.ParseFromString(proto_buf))
						LOG_ERROR_VA("[%s] gf_match_resp ParseFromString fail", get_name());
					else
						gf_match_resp_vec.push_back(gf_match_resp);
					alive_gf_match_vec.erase(alive_gf_match_vec.begin() + index);
					index--;
					recv_match_req_num++;
					zmq_msg_close(&gf_match_resp_msg);
				}
				index++;
			}
		}
		//recv all match result sucess
		//result in gf_match_resp_msg_vec
		//start Merge
		
		int resp_num = gf_match_resp_vec.size();
		LOG_DEBUG_VA("[%s] recv %d gf_match_resp", get_name(), resp_num);
		GLOBALFEATURE_3_2_0::TDistributedResult *res = new GLOBALFEATURE_3_2_0::TDistributedResult[resp_num];
		GLOBALFEATURE_3_2_0::TDistributedResult *tmp_res = NULL;
		GLOBALFEATURE_3_2_0::TDRes *tmp_gf = NULL;
		GLOBALFEATURE_3_2_0::TDRes *tmp_cf = NULL;

		for(int i = 0; i < resp_num; i++){
			//the input of merge
			tmp_res = (GLOBALFEATURE_3_2_0::TDistributedResult *)(gf_match_resp_vec[i].res().c_str());
			tmp_gf = (GLOBALFEATURE_3_2_0::TDRes *)(gf_match_resp_vec[i].gf_res().c_str());
			tmp_cf = (GLOBALFEATURE_3_2_0::TDRes *)(gf_match_resp_vec[i].cf_res().c_str());
			res[i].nResults = tmp_res->nResults;
			res[i].pImgRes = tmp_gf;
			res[i].pClrRes = tmp_cf;
			for(int j = 0; j < 5; j++){
				LOG_DEBUG_VA("[%s] get match result iid: %llu, gid: %llu", get_name(),
					res[i].pImgRes[j].imageID, res[i].pImgRes[j].groupID);
			}
			for(int j = 0; j < 5; j++){
				res[i].min_dist[i] = tmp_res->min_dist[i];
				res[i].max_dist[i] = tmp_res->max_dist[i];
			}
			res[i].imageType = tmp_res->imageType;
			//input of merge over
		}//end of gf_match_resp_vec

		//output of merge
		int gf_num = gf_match_req.gf_num();
		int cf_num = gf_match_req.cf_num();
		GLOBALFEATURE_3_2_0::TImageRes *global_res = new GLOBALFEATURE_3_2_0::TImageRes[gf_num];
		GLOBALFEATURE_3_2_0::TImageRes *color_res = new GLOBALFEATURE_3_2_0::TImageRes[cf_num];
		//output of merge over

		ret = GLOBALFEATURE_3_2_0::Merge(gf_match_req.class_id(), resp_num, res, gf_match_req.merge_same(), gf_num, global_res, cf_num, color_res);
		if(ret != 0){
			LOG_ERROR_VA("[global] Merge failed: %d", ret);
			send_error_reply(receiver);
			delete [] res;
			delete [] global_res;
			delete [] color_res;
			continue;
		}
		delete [] res;
		res = NULL;
		//end of merge
		//start send resp to front
		LOG_DEBUG_VA("[%s] prepare to send gf_merge_resp to front", get_name());
		proto::gf_merge_resp gf_merge_resp;
		gf_merge_resp.mutable_head()->set_version(420);
		gf_merge_resp.mutable_head()->set_command(420);
		gf_merge_resp.mutable_head()->set_password("xxxxxxxx", 8);
		gf_merge_resp.mutable_head()->set_type(1);
		gf_merge_resp.mutable_head()->set_status(0);
		gf_merge_resp.mutable_head()->set_src(1);
		gf_merge_resp.mutable_head()->set_dest(2);
		gf_merge_resp.mutable_head()->set_sn_high(123456789);
		gf_merge_resp.mutable_head()->set_sn_low(123456789);
		gf_merge_resp.mutable_head()->set_length(0);

		gf_merge_resp.set_res_num(gf_num + cf_num);
		
		for(int i = 0; i < gf_num; i++){
			gf_merge_resp.add_iid(global_res[i].imageID);
			gf_merge_resp.add_gid(global_res[i].groupID);
			gf_merge_resp.add_simi(global_res[i].similarity);
			gf_merge_resp.add_type(global_res[i].imageType);
			gf_merge_resp.add_source(global_res[i].source);
		}
		for(int i = 0; i < cf_num; i++){
			gf_merge_resp.add_iid(color_res[i].imageID);
			gf_merge_resp.add_gid(color_res[i].groupID);
			gf_merge_resp.add_simi(color_res[i].similarity);
			gf_merge_resp.add_type(color_res[i].imageType);
			gf_merge_resp.add_source(color_res[i].source);
		}
		delete [] global_res;
		delete [] color_res;
		global_res = NULL;
		color_res = NULL;
		//merge over
		proto_buf.clear();
		if(!gf_merge_resp.SerializePartialToString(&proto_buf)){
			LOG_ERROR_VA("SerializePartialToString fail");
			send_error_reply(receiver);
			continue;
		}
		zmq_msg_t gf_merge_resp_msg;
		zmq_msg_init_size(&gf_merge_resp_msg, proto_buf.size());
		memcpy(zmq_msg_data(&gf_merge_resp_msg), proto_buf.c_str(), proto_buf.size());
		LOG_DEBUG_VA("[%s] send gf_merge_resp to front", get_name());
		size = zmq_sendmsg(receiver, &gf_merge_resp_msg, 0);
		if(size == -1){
			send_error_reply(receiver);
			LOG_ERROR_VA("zmq_sendmsg error");
			continue;
		}
		for(int i = 0; i < 5; i++){
			LOG_DEBUG_VA("[%s] result iid: %llu, gid: %llu, simi: %d, source: %d", get_name(),
				gf_merge_resp.iid(i), gf_merge_resp.gid(i), gf_merge_resp.simi(i), gf_merge_resp.source(i));
		}
	}
	zmq_close(receiver);
}

int gf_match_thread::thread_func(){
	if(get_thread_pool() == NULL){
		LOG_ERROR_VA("[%s] pool_work_thread don't belong to a thread_pool", get_name());
		return -1;
	}
	int ret;
	void *receiver = zmq_socket(get_thread_pool()->get_zmq(), ZMQ_REP);
	if(receiver == NULL){
		LOG_ERROR_VA("[%s] zmq_socket failed", get_name());
		return -1;
	}
	//ret = zmq_connect(receiver, "inproc://workers");
	ret = zmq_connect(receiver, "tcp://199.155.122.11:45513"); //gf_match back_address

	LOG_DEBUG_VA("[%s] thread start...", get_name());
	while(!(get_thread_pool()->has_stoped())){
		int size;
		zmq_msg_t recv_msg;
		zmq_msg_init(&recv_msg);

		LOG_DEBUG_VA("[%s] thread get task...", get_name());
		size = zmq_recvmsg(receiver, &recv_msg, 0);
		if (size == -1) {
			LOG_ERROR_VA("zmq_recvmsg error");
			send_error_reply(receiver);
			continue;
		}
		string proto_buf;
		proto_buf.assign((char *)zmq_msg_data(&recv_msg), size);
		proto::gf_match_req gf_match_req;
		if(!gf_match_req.ParseFromString(proto_buf)){
			LOG_ERROR_VA("[%s] gf_match_req ParseFromString fail", get_name());
			send_error_reply(receiver);
			continue;
		}
		LOG_DEBUG_VA("[%s] recv length: %d", get_name(), size);
		//start Match
		database_bdb *db = get_database();

		int vid_num = gf_match_req.vids_size();
		int *vids = new int[vid_num];
		if(vid_num != 8)
			LOG_ERROR_VA("vid_num != 8, %d", vid_num);

		for(int i = 0; i < vid_num; i++){
			vids[i] = gf_match_req.vids(i);
		}

		vid_data *vid_datas = new vid_data[vid_num];
		if(db->get_data(gf_match_req.class_id(), gf_match_req.b2c_flag(), database::gf, vid_num, vids, vid_datas)  <= 0){
			LOG_ERROR_VA("[gf_match] db get data failed, check data_root_path config");
			send_error_reply(receiver);
			delete [] vid_datas;
			delete [] vids;
			continue;
		}
		GLOBALFEATURE_3_2_0::TDataStruct *datas = new GLOBALFEATURE_3_2_0::TDataStruct[vid_num];
		for (int i = 0; i != vid_num; i++) {
			datas[i].dataLen = vid_datas[i].length;
			datas[i].data = vid_datas[i].data;
		}
		delete [] vid_datas;
		GLOBALFEATURE_3_2_0::TDistributedResult *res = new GLOBALFEATURE_3_2_0::TDistributedResult;
		
		res->nResults = gf_match_req.gf_num() > gf_match_req.cf_num() ? gf_match_req.gf_num() : gf_match_req.cf_num();
		res->pImgRes = new GLOBALFEATURE_3_2_0::TDRes[res->nResults];
		res->pClrRes = new GLOBALFEATURE_3_2_0::TDRes[res->nResults];
		res->imageType = 0;

		char *feature_buf = new char[gf_match_req.feature_data().size()];
		memcpy(feature_buf, gf_match_req.feature_data().c_str(), gf_match_req.feature_data().size());

		ret = GLOBALFEATURE_3_2_0::Match(vid_num, datas, feature_buf, gf_match_req.merge_same(), res); 
		if(ret != 0){
			LOG_ERROR_VA("Global Match failed: %d", ret);
			send_error_reply(receiver);
			for (int i = 0; i != vid_num; i++)
				if (datas[i].dataLen)
					delete[] datas[i].data;
			delete [] datas;
			delete [] res->pImgRes;
			delete [] res->pClrRes;
			continue;
		}
		for (int i = 0; i != vid_num; i++)
			if (datas[i].dataLen)
				delete[] datas[i].data;
		delete [] datas;
		for(int i = 0; i < 5; i++){
			LOG_DEBUG_VA("[%s] result iid: %llu, gid: %llu", get_name(),
				res->pImgRes[i].imageID, res->pImgRes[i].groupID);
		}
	
		//end of match
		//result in GLOBALFEATURE_3_2_0::TDistributedResult res;
		proto::gf_match_resp gf_match_resp;
		gf_match_resp.mutable_head()->set_version(420);
		gf_match_resp.mutable_head()->set_command(420);
		gf_match_resp.mutable_head()->set_password("xxxxxxxx", 8);
		gf_match_resp.mutable_head()->set_type(1);
		gf_match_resp.mutable_head()->set_status(0);
		gf_match_resp.mutable_head()->set_src(1);
		gf_match_resp.mutable_head()->set_dest(2);
		gf_match_resp.mutable_head()->set_sn_high(123456789);
		gf_match_resp.mutable_head()->set_sn_low(123456789);
		gf_match_resp.mutable_head()->set_length(0);

		gf_match_resp.set_gf_res(res->pImgRes, sizeof(GLOBALFEATURE_3_2_0::TDRes) * res->nResults);
		gf_match_resp.set_cf_res(res->pClrRes, sizeof(GLOBALFEATURE_3_2_0::TDRes) * res->nResults);

		delete [] res->pImgRes;
		delete [] res->pClrRes;
		res->pImgRes = NULL;
		res->pClrRes = NULL;
		gf_match_resp.set_res(res, sizeof(GLOBALFEATURE_3_2_0::TDistributedResult));
		delete [] res;
		res = NULL;
		proto_buf.clear();
		if(!gf_match_resp.SerializePartialToString(&proto_buf)){
			LOG_ERROR_VA("SerializePartialToString fail");
			send_error_reply(receiver);
			continue;
		}

		zmq_msg_t match_resp_msg;
		zmq_msg_init_size(&match_resp_msg, proto_buf.size());
		memcpy(zmq_msg_data(&match_resp_msg), proto_buf.c_str(), proto_buf.size());
		size = zmq_sendmsg(receiver, &match_resp_msg, 0);
		if(size == -1){
			send_error_reply(receiver);
			LOG_ERROR_VA("zmq_sendmsg error");
			continue;
		}
	}
	zmq_close(receiver);
}

database_bdb *gf_match_thread::get_database(){
	match_server *server =dynamic_cast<match_server *>(get_thread_pool()->get_server());
	return server->get_database();
}

vector<address> &gf_merge_thread::get_gf_match(){
	gf_merge_server *server =dynamic_cast<gf_merge_server *>(get_thread_pool()->get_server());
	return server->get_gf_match();
}


